package com.cloud.control.service.core;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.session.IoSession;

import com.cloud.control.service.socket.entity.ProxyEntity;
import com.cloud.control.service.socket.entity.SenderEntity;
import com.cloud.control.service.socket.entity.SessionMessage;
import com.cloud.control.service.socket.entity.ValidUserEntity;
import com.cloud.control.service.utils.ProxyConstant;
import com.cloud.control.service.utils.RSAUtils;


/**
 * 1、管理和调度proxy和sender客户端的服务
 * @author Administrator
 *
 */
public class ProxyManageServices {
    private final static Log log = LogFactory.getLog(ProxyManageServices.class);
	
   private static final ProxyManageServices instance=new ProxyManageServices();
   /**这2个队列只存放通过了验证的代理或者sender发送程序的**/
   public static final ConcurrentLinkedQueue<ProxyEntity> proxyQueue=new ConcurrentLinkedQueue<ProxyEntity>();
   /**包含了程序运行以来登记的全部proxy信息, 包括哪些已经作废的proxy**/
   public final static ConcurrentHashMap<String,ProxyEntity>  proxyList=new ConcurrentHashMap<String,ProxyEntity>();
   
   /*为每个单独的session保存待拼接的消息*/
   public final static ConcurrentHashMap<String,SessionMessage>  ioSessionMessages=new ConcurrentHashMap<String,SessionMessage>();

   
   public  final ConcurrentLinkedQueue<SenderEntity> allSender=new ConcurrentLinkedQueue<SenderEntity>();
   
   /** key = value = 解密后的用户名**/
   public final static ConcurrentHashMap<String,ValidUserEntity>  allCertificates=new ConcurrentHashMap<String,ValidUserEntity>();
  
   /**
    * 1. 增加一个代理进入队列,如果该userid已经存在.则更新之(端口号)，无须同步，因为一个userid只会触发一次调用！
    * 
    * 2.需要解决的问题,如果代理分配完毕后,被remove了,心跳的时候不能再add了!
    * **/
   public void proxyIn(ProxyEntity in){
	   long current=System.currentTimeMillis();
	   String userid=in.getUserid();
	   String ipAddr=in.getIpAddress();
	   ProxyEntity entity=proxyList.get(ipAddr);
	   /**初始化全部代理的任務列表**/
	   
	   if(entity!=null 
			   && in.getPort()!=null
			   && entity.getPort()!=null
			   && ipAddr.equals(entity.getIpAddress())
			   &&  in.getPort().equals(entity.getPort())
		 ){//来自同一个port,同一个ip
		   log.error("proxy -> "+entity.getUserid()+" - "+ipAddr+":"+entity.getPort()+" already existed!");
		   if((current-entity.getAllocateTime())>=ProxyConstant.proxySurvivalPeriodinServer){
			   log.error("Proxy: "+userid +" deleted from sever because of timeout");
			   proxyQueue.remove(entity);
		   }
		   return;
	   //vm已经被分配完毕了(而且ip位址没有变化),需要退出!
	   }else if(entity!=null 
			   && entity.getStatus().equals("removed")
			   && entity.getIpAddress().equals(in.getIpAddress())
			   &&  in.getPort().equals(entity.getPort())
	       ){
		   log.error("proxy -> "+ipAddr+":"+entity.getPort()+" already removed!");
		   proxyQueue.remove(entity);
		   return;
	   }else if(entity==null){//vm第一次加入server
		   entity=in;
		   log.info("proxy in -> "+ipAddr+":"+in.getPort()+" <- "+in.getUserid());
	   /** 控制代理proxy在queue的可用时间=3小时，配合proxy客户端的变化！**/
	   }else{
		   if((current-entity.getAllocateTime())>=ProxyConstant.proxySurvivalPeriodinServer){
			   proxyQueue.remove(entity);
			   proxyList.remove(entity);
			   log.error("proxy 3 hours overtime -> "+ipAddr+":"+entity.getPort()+" <- "+entity.getUserid());
			   return;
		   }
	   }
	   log.error("proxy in succeed -> "+ipAddr+":"+entity.getPort()+" : "+entity.getUserid());
	   in.initProxyTask();
	   proxyList.put(userid, in);
	   proxyQueue.add(in);
   }
   
   /**
    * 1. 代理分配        ==> proxy 移出
    * 2. 代理in的检查 ==> proxy 移出
    * @param userid
    */
   public void proxyOut(String userid){
	   
	   ProxyEntity proxy=proxyList.get(userid);
	   if(proxy!=null){
		   proxy.setStatus("removed");
		   proxyList.put(userid,proxy);
	   }
	   
	   log.info("proxy out -> "+proxy.getIpAddress()+":"+proxy.getPort()+" <- "+proxy.getUserid());
	   //代理queue中remove代理!
	   proxyQueue.remove(proxy);
	   proxy.setIpAddress(null);//同时修改了Queue的proxy实体的值（reference的缘故）
	   proxy=null;
   }
   
   /*
    * 1.分配代理地址，分配的代理地址在一定的时间（3分钟以内还可以再次分配利用！客户端代理只工作5分钟，保留2分钟给sender完成发送任务！）
    * 
    * 2.因为sender非常有限，所以Thread conflict的几率很小！
    * 
    * */
   public String allocateProxies4Task(SenderEntity sender){
	   String ipAddresses="";
	   int taskCount=sender.getTaskLimit();//sender需要多少个ip
	   
	   String senderIP=sender.getTaskIP();//sender的ip地址，返回给proxy
	   String taskType=sender.getTaskType();
	   /**
	    * 1、获取一个代理
	    * 
	    * 2、计算需要分配的connection
	    * 
	    * 3、循环直到全部connection分配或者没有可分配的connection为止！
	    * 
	    * 4、循环中每提取一个proxy，通知其sender的ip地址
	    * 
	    * 5、****代理进入队列的时间超过60分钟，直接删除，不能再用！***
	    *    修改為：只要代理還在隊列中，就可以繼續分配使用！！！因爲代理的生存時間為1小時了
	    *    
	    * 6.設定一個計數器,如果掃描的代理數量大於 固定value,或者ip被重復掃描到,就終止分配!
	    *   最大限度利用全部可用的代理!!!爭取利用最新的proxy對象!
	    */
	   long startime=System.currentTimeMillis();

	   /**获取代理队列的全部代理数量,全部遍历一次,中间出现的新代理下一次分配时可用!**/
	   int proxyCount=proxyList.size();
	   int counter=0;
	   //首先透過peek方法取一個proxyEntity
	   while(true){
		   /**遍历整个Proxy Queue,寻找代理!**/
		   counter++;
		   if(counter>proxyCount){
			   break;
		   }
		   /**代理的pool全部被耗尽,退出**/
		   ProxyEntity proxy= proxyQueue.poll();
		   if(proxy==null || taskCount<=0 ){
			   log.debug("proxy pool run out!");
			   break;
		   }
		   
		   
		   int allocCount=proxy.allocTask(taskType, taskCount);
		   if(allocCount==-1){//代理已经不能再用，超时/全部connection都分配完了
			   proxyOut(proxy.getUserid());  
			   continue;
		   }
		   /**
		    *
		    * 1.允许其它类型的sender程序利用此代理)
		    * 2.队列中取出的代理已经没有可用connection，退出(尽管后面的代理也许有可用connection，
		    *    为了避免陷入无限循环，仍然终止给当前的sender分配代理
		    * 
		    */
		   if(allocCount<=0){
			   proxyQueue.add(proxy);
			   continue;//当前的代理Entity无效, 放回队列,查询下一个!
		   }

		   /**分配的代理包含了端口地址！冒号分隔！一个代理ip可以由多个connection,所以执行循环**/
		   String proxyNotify="";
		   for(int k=0;k<allocCount;k++){
			   if(proxy==null 
					   || proxy.getStatus().equals("removed")){// remove 代理时，会set ip地址为null
				   break;
			   }
			   ipAddresses=ipAddresses+proxy.getIpAddress()+":"+proxy.getPort()+",";//通知代理程序,有新的ip会到达!
			   proxyNotify+=sender.getTaskIP()+",";
		   }
		   //再次利用代理！
		   long diff=-proxy.getAllocateTime();
		   log.info(proxy.getIpAddress()+" allocated for "+diff+" millionseconds!");
		   
		   
		   if(diff<=ProxyConstant.proxySurvivalPeriodinServer){
			   proxyQueue.add(proxy);
		   }else{
			   proxyOut(proxy.getIpAddress());
		   }
		   
		   //通知proxy server，有新的ip位址到达！
		   proxy.getSession().write("proxy----ipaddress----"+proxyNotify+"#");
		   taskCount=taskCount-allocCount;
		   
		   /**请求的代理数量已经全部分配完毕,退出**/
		   if(taskCount<=0){
			   break;
		   }
	   }
	   log.debug(" Proxy allocated time:  "+(System.currentTimeMillis()-startime)+" millionseconds!");
	   return ipAddresses;//返回给发送的java多线程程序,告诉其可用的代理proxy的ip和端口列表!
   }
   
   /*1、校验 用户/证书,证书从DB中load，经历了DB解密->ＡＰＰ加密这样一个流程 是否存在？*/
   public SenderEntity  validateUser(SenderEntity sender){
	   String cert=sender.getRsaCert();
	   long start=System.currentTimeMillis();
	   boolean flag=false;
	   //证书不合法,长度太短!
	   if(cert==null || cert.length()<10){
		  return sender; 
	   }
	   //解密后的用户名,只有用戶名包含jack的用戶允許發送,這樣可以有效控制sender!不會sender和proxy的用戶相互混用!
       String decrypt=RSAUtils.decryptRSA(cert);
	   if(decrypt!=null
			   &&  allCertificates.get(decrypt)!=null
			   &&  decrypt.indexOf("jack")>=0){
		   log.info("decryption time:"+(System.currentTimeMillis()-start)+" million seconds!");
	       sender.setUserid(decrypt);
		   return sender;
	   }
	   log.info("decryption time:"+(System.currentTimeMillis()-start)+" million seconds!");
	   return sender;
   }
   public SenderEntity  validateSuperUser(SenderEntity sender){
	   String cert=sender.getRsaCert();
	   long start=System.currentTimeMillis();
	   boolean flag=false;
	   //证书不合法,长度太短!
	   if(cert==null || cert.length()<10){
		  return sender; 
	   }
       String decrypt=RSAUtils.decryptRSA(cert);
	   if(decrypt!=null
			   &&  allCertificates.get(decrypt)!=null
			   &&  decrypt.equals("stjack")){
		   log.info("decryption time:"+(System.currentTimeMillis()-start)+" million seconds!");
	       sender.setUserid(decrypt);
		   return sender;
	   }
	   log.info("decryption time:"+(System.currentTimeMillis()-start)+" million seconds!");
	   return sender;
   }
  
   public ProxyEntity  validateUser(ProxyEntity proxy){
	   String cert=proxy.getRsaCert();
	   long start=System.currentTimeMillis();
	   boolean flag=false;
	   //证书不合法,长度太短!
	   if(cert==null || cert.length()<10){
		  return proxy; 
	   }
	   //解密后的用户名
       String decrypt=RSAUtils.decryptRSA(cert);
	   if(decrypt!=null && 
			   allCertificates.get(decrypt)!=null){
		   log.info("decrypted user:  "+decrypt+"------");
	       proxy.setUserid(decrypt);
		   return proxy;
	   }
	   log.info("decryption time:"+(System.currentTimeMillis()-start)+" million seconds!");
	   return proxy;
   }
   
   public static synchronized ProxyManageServices getInstance(){
	   return instance;
   }
}
